/*
 * delta_phase.c
 *
 *  Created on: Jul 28, 2014
 *      Author: fumin
 */
#include "destor.h"
#include "jcr.h"
#include "backup.h"
#include "storage/containerstore.h"
#include <openssl/sha.h>
#include <openssl/md5.h>
#include "storage/containerstore.h"
#include "xdelta/xdelta3.h"
#include "utils/lru_cache.h"

#define MAXCHUNKSIZE 4096*64

#define DELTASCHEME 1 //0 dDelta, 1 xDelta, 2 zdelta
#define DELTACHECK  0  //check for restore
#define DELTASTAGE4 0

#define SLIDE1(m,fp,bufPos,buf) do{	\
	    unsigned char om;   \
	    u_int64_t x;	 \
		if (++bufPos >= size)  \
            bufPos = 0;				\
        om = buf[bufPos];		\
        buf[bufPos] = m;		 \
		fp ^= U[om];	 \
		x = fp >> shift;  \
		fp <<= 8;		   \
		fp |= m;		  \
		fp ^= T[x];	 \
}while(0)

typedef unsigned int uint32_t;
typedef unsigned char uint8_t;
typedef unsigned short uint16_t;
typedef unsigned int u_int32_t;

typedef unsigned long long int UINT64;

extern struct{
	/* accessed in dedup phase */
	struct container *container_buffer;
	/* In order to facilitate sampling in container,
	 * we keep a queue for chunks in container buffer. */
	GQueue *chunks;
} storage_buffer;

static pthread_t delta_t;

//static GHashTable *delta_index;

/*containerid lookup_delta_index(fingerprint *fp){
	pthread_mutex_lock(&mutex);
	struct base_chunk_pointer* p = g_hash_table_lookup(delta_index, fp);
	if(p){
		pthread_mutex_unlock(&mutex);
		return p->id;
	}
	pthread_mutex_unlock(&mutex);
	NOTICE("Fail delta: The base chunk has not been written to a container yet!");
	return TEMPORARY_ID;
}
*/

int ComputeDelta(struct chunk* basChunk, struct chunk* srcChunk)
{
	if(basChunk->size < 1024 || srcChunk->size < 512){
        srcChunk->delta = NULL;
		return 0;
	}

	unsigned char deltaData[128*1024];//[128*1024];
	uint32_t dSize;

#if DELTASCHEME == 1
	//========xDelta encoding...
	//ComJob.SimilDataSize += srcChunk->nChunkSize;
	//gettimeofday(&ComJob.deltaStart, NULL);
	int ret = xd3_encode_memory((uint8_t*)srcChunk->data, (usize_t)srcChunk->size,
								(uint8_t*)basChunk->data, (usize_t)basChunk->size,
								(uint8_t*)deltaData, (usize_t*)&dSize, 
								(usize_t)srcChunk->size, 0);
	/*gettimeofday(&ComJob.deltaEND, NULL);
	ComJob.tEncoding += (ComJob.deltaEND.tv_sec - ComJob.deltaStart.tv_sec)*1000 + 
						(ComJob.deltaEND.tv_usec - ComJob.deltaStart.tv_usec)/1000.0;
    */
	if(ret != 0){
		return 0;
	}
#if DELTACHECK
	u_int32_t rsize;
	unsigned char restore[MAXCHUNKSIZE];
	//gettimeofday(&ComJob.deltaStart, NULL);
	int r2 = xd3_decode_memory( (uint8_t*)deltaData, (usize_t)dSize,
					( uint8_t*)basChunk->data, (usize_t)basChunk->size,
					( uint8_t*)restore, (usize_t*)&rsize,(usize_t)srcChunk->size, 1);
	/*gettimeofday(&ComJob.deltaEND, NULL);
	ComJob.tDecoding += (ComJob.deltaEND.tv_sec - ComJob.deltaStart.tv_sec)*1000 + 
						(ComJob.deltaEND.tv_usec - ComJob.deltaStart.tv_usec)/1000.0;
    */
	//if(rsize != srcChunk->size)
	//	exit(1);

	if(memcmp(srcChunk->data, restore, rsize) != 0){
		printf("xDelta decode error!!!\n");
		assert(memcmp(srcChunk->data, restore, rsize) == 0);
	}
#endif
#endif

	if(dSize > srcChunk->size) {
        srcChunk->delta = NULL;
        return 0;
	}

    srcChunk->delta = (struct delta*)malloc(sizeof(struct delta));
    
    memcpy(srcChunk->delta->basefp, basChunk->fp, sizeof(fingerprint));
    srcChunk->delta->baseid = basChunk->id;
    srcChunk->delta->base_size = basChunk->size;
    srcChunk->delta->size = dSize;
    srcChunk->delta->data = (unsigned char *)malloc(dSize);
	memcpy(srcChunk->delta->data, deltaData, dSize);
    
	return 1;
}

void *delta_thread(void *arg)
{
    while (1) {
        struct chunk* c = sync_queue_pop(matching_queue);

        if (c == NULL)
            /* backup job finish */
            break;

        if(CHECK_CHUNK(c, CHUNK_FILE_START) ||
        		CHECK_CHUNK(c, CHUNK_FILE_END) ||
        		CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
        		CHECK_CHUNK(c, CHUNK_SEGMENT_END)){
        	sync_queue_push(delta_queue, c);
        	continue;
        }

        if(CHECK_CHUNK(c, CHUNK_DUPLICATE) || c->size <= 2048){
        	sync_queue_push(delta_queue, c);
        	continue;
        }

        if (c->baseChunk)
            if (ComputeDelta(c->baseChunk, c)) {
                jcr.number_of_chunks_has_delta++;

                int compressed_size = c->size - c->delta->size;
                jcr.total_compressed_size += compressed_size;
				jcr.total_size_for_deltacompression += c->size;
            }
            else {
                c->delta = NULL;
                free_chunk(c->baseChunk);
                c->baseChunk = NULL;
                jcr.unproper_delta_comp_num++;
            }
        
        sync_queue_push(delta_queue, c);
    }

    sync_queue_term(delta_queue);

	return NULL;
}

void start_delta_phase() {
    
    delta_queue = sync_queue_new(1000);
    pthread_create(&delta_t, NULL, delta_thread, NULL);
}

void stop_delta_phase() {
    pthread_join(delta_t, NULL);
}
